package com.wyh.apitest.transformation.multithread

import com.wyh.apitest.source.SensorReading
import org.apache.flink.streaming.api.scala._

object ConnectCoMap {
  def main(args: Array[String]): Unit = {
    //定义流式处理环境
    val streamEnv = StreamExecutionEnvironment.getExecutionEnvironment
    streamEnv.setParallelism(1)

    val sensorinfo: DataStream[String] = streamEnv.readTextFile("D:\\IdeaProjects\\FlinkStudyScala\\src\\main\\resources\\sensor.txt")
    //转换成样例类类型
    val dataStream = sensorinfo.map(s => {
      val arr: Array[String] = s.split(",")
      SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
    })

    //多流转换算子----分流操作(在ProessFunction的代码里讲解另一种解法侧输出流)
    val splitStream: SplitStream[SensorReading] = dataStream.split(data => {
      if (data.temperature > 30.0) {
        Seq("higth")
      } else {
        Seq("low")
      }
    })

    //多流转换算子----合流操作
    //挑选出温度大于30的
    val higthTempStream: DataStream[SensorReading] = splitStream.select("higth")
    //挑选出温度小于30的
    val lowTempStream: DataStream[SensorReading] = splitStream.select("low")
    //转化一下与其他不一样的格式
    val warnTempStream: DataStream[(String, Double)] = higthTempStream.map(data => {
      (data.id, data.temperature)
    })
    //DataStream,DataStream → ConnectedStreams
    //connect只能连接两条流
    val connectedStream: ConnectedStreams[(String, Double), SensorReading] = warnTempStream.connect(lowTempStream)
    //用CoMap分别处理
    //ConnectedStreams → DataStream
    val coMapResultStream: DataStream[Product with Serializable] = connectedStream
      .map(
        waringData => (waringData._1, waringData._2, "高温预警"),
        lowTempData => (lowTempData.id, "正常")
      )

    coMapResultStream.print("CoMap")


    streamEnv.execute("多流转换算子，ConnectCoMap")

  }

}